package com.jjb.spider.mqtt.handler;

import com.google.common.base.Strings;
import com.jjb.spider.mqtt.cache.NodeChannelHandlerContextCache;
import com.jjb.spider.mqtt.ctx.MqttChannelHandlerContext;
import com.jjb.spider.mqtt.service.IMqttMessageService;
import com.jjb.spider.mqtt.service.LoginService;
import com.jjb.spider.mqtt.service.WillService;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.mqtt.*;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.util.AttributeKey;
import io.netty.util.concurrent.Future;
import io.netty.util.concurrent.GenericFutureListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

@Component
public class MqttHandler {

    private Logger log = LoggerFactory.getLogger(MqttHandler.class);

    @Autowired
    private LoginService loginService;

    @Autowired
    private IMqttMessageService mqttMessageService;

    @Autowired
    private WillService willService;

    private MqttChannelHandlerContext getMqttChannelHandlerContext(ChannelHandlerContext context) {
        return NodeChannelHandlerContextCache.getByChannel(context);
    }

    public void mqttHandler(ChannelHandlerContext ctx, MqttMessage mqttMessage) {
        MqttFixedHeader mqttFixedHeader = mqttMessage.fixedHeader();

        if( mqttFixedHeader == null ) {
            MqttChannelHandlerContext mqctx = getMqttChannelHandlerContext(ctx);
            String deviceId=null;
            if( mqctx != null ){
                deviceId = mqctx.getDeviceId();
            }
            log.info("请求协议头不完整:{},deviceId:{}",ctx.channel().remoteAddress().toString(),deviceId);
            ctx.close().addListener(new GenericFutureListener<Future<? super Void>>() {
                @Override
                public void operationComplete(Future<? super Void> future) throws Exception {
                    if( future.isSuccess() ) {
                        log.info("连接已关闭");
                    } else {
                        log.info("连接关闭失败");
                    }
                }
            });
            return;
        }

        if( mqttFixedHeader.messageType() == MqttMessageType.CONNECT ) {
            //当前为连接请求，处理登录信息
            this.connection(ctx, (MqttConnectMessage)mqttMessage);
            return;
        }

        MqttChannelHandlerContext mqttChannelHandlerContext = getMqttChannelHandlerContext(ctx);

        if (! mqttChannelHandlerContext.isLogin() ) {
            //当前未登录，直接关闭连接
            ctx.close();
            return;
        }

        //开始处理
        switch (mqttFixedHeader.messageType()) {
            case PINGREQ:  //心跳请求
                log.debug("收到心跳请求");
                pingRes(mqttChannelHandlerContext);
                break;
            case SUBSCRIBE: //订阅主体
                log.debug("收到订阅请求");
                subscribe(mqttChannelHandlerContext, (MqttSubscribeMessage) mqttMessage);
                break;
            case UNSUBSCRIBE: //取消主体订阅
                log.debug("收到取消订阅请求");
                unsubscribe(mqttChannelHandlerContext, (MqttUnsubscribeMessage) mqttMessage);
                break;
            case DISCONNECT:   //客户端主动断开连接
                log.debug("收到断开连接请求");
                this.closeLogin(mqttChannelHandlerContext);
                ctx.close();
                break;
            case PUBLISH:
                log.debug("收到发布请求");
                publish(mqttChannelHandlerContext, (MqttPublishMessage) mqttMessage);
                break;
            case PUBACK:
                log.debug("收到消息确认请求");
                //pubback(mqttChannelHandlerContext, mqttMessage);
                break;
            case PUBREC:
                pubrec(mqttChannelHandlerContext,  mqttMessage);
                break;
            case PUBREL:
                pubrel(mqttChannelHandlerContext,  mqttMessage);
                break;
            case PUBCOMP:
                pubcomp(mqttChannelHandlerContext,  mqttMessage);
                break;
            default:
                break;
        }
    }



    private void pubcomp(MqttChannelHandlerContext mqttChannelHandlerContext, MqttMessage mqttMessage) {
        this.mqttMessageService.pubcomp(mqttChannelHandlerContext, mqttMessage).subscribe();
    }

    private void pubrel(MqttChannelHandlerContext mqttChannelHandlerContext, MqttMessage mqttMessage) {
        this.mqttMessageService.pubrel(mqttChannelHandlerContext, mqttMessage).subscribe();
    }

    private void pubrec(MqttChannelHandlerContext mqttChannelHandlerContext, MqttMessage mqttMessage) {
        this.mqttMessageService.pubrec(mqttChannelHandlerContext, mqttMessage).subscribe();
    }

    private void pubback(MqttChannelHandlerContext ctx, MqttMessage mqttMessage) {
        System.out.println("收到确认消息节点");
        this.mqttMessageService.pubback(ctx, (MqttPubAckMessage) mqttMessage).subscribe();
    }


    private void publish(MqttChannelHandlerContext ctx, MqttPublishMessage mqttMessage) {
        this.mqttMessageService
                .publish(ctx, mqttMessage)
                .doOnError(throwable -> log.error("执行失败:" + throwable.getMessage(), throwable))
                .subscribe();
    }

    private void unsubscribe(MqttChannelHandlerContext ctx, MqttUnsubscribeMessage mqttMessage) {
        this.mqttMessageService.unsubscribe(ctx, mqttMessage).subscribe();
    }

    private void subscribe(MqttChannelHandlerContext ctx, MqttSubscribeMessage mqttMessage) {
        this.mqttMessageService.subscribe(ctx, mqttMessage).subscribe();
    }

    /**
     * 心跳请求回复
     * @param ctx
     */
    private void pingRes(MqttChannelHandlerContext ctx) {
        MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.PINGRESP, false, MqttQoS.AT_MOST_ONCE, false, 0);
        MqttMessage mqttMessageBack = new MqttMessage(fixedHeader);
        ctx.writeAndFlush(mqttMessageBack);
    }

    /**
     * 释放连接
     * @param context
     */
    private void closeLogin(MqttChannelHandlerContext context) {
        context.getContext().attr(_login).set(false);
        context.getContext().attr(_deviceId).set("");

        NodeChannelHandlerContextCache.delChannelHandlerContext(context);
    }

    private void connection(ChannelHandlerContext ctx, MqttConnectMessage mqttMessage) {

        MqttConnectPayload payload = mqttMessage.payload();
        String deviceId = payload.clientIdentifier();
        if (Strings.isNullOrEmpty(deviceId)) {
            //未获取到设备ID， 直接结束
            MqttConnectReturnCode connectReturnCode = MqttConnectReturnCode.CONNECTION_REFUSED_IDENTIFIER_REJECTED;
            connectBack(ctx, connectReturnCode);
            return;
        }

        String username = null, password = null;
        if( mqttMessage.variableHeader().hasUserName() ) {
            username = payload.userName();
        }

        if( mqttMessage.variableHeader().hasPassword() ) {
            password = new String(payload.passwordInBytes());
        }

        this.loginService.login(deviceId, username, password)
            .map(bool -> {
                if (bool) {
                    loginSuccess(ctx, deviceId);
                    //登录成功绑定will标签
                    this.willService.loginSuccess(deviceId,  ctx,  mqttMessage).subscribe();
                    MqttConnectReturnCode connectReturnCode = MqttConnectReturnCode.CONNECTION_ACCEPTED;
                    return connectReturnCode;
                } else {
                    MqttConnectReturnCode connectReturnCode = MqttConnectReturnCode.CONNECTION_REFUSED_BAD_USER_NAME_OR_PASSWORD;
                    return connectReturnCode;
                }
            }).doOnError(throwable -> {
                log.error("客户端创建连接失败，失败原因:" + throwable.getMessage(), throwable);
                connectBack(ctx, MqttConnectReturnCode.CONNECTION_REFUSED_PROTOCOL_ERROR);
            })
            .subscribe(mqttConnectReturnCode -> {
                connectBack(ctx, mqttConnectReturnCode);
                log.info("mqtt连接成功ip:{},deviceId:{}",ctx.channel().remoteAddress().toString(),deviceId);
            });
    }

    protected AttributeKey<Boolean> _login = AttributeKey.valueOf("login");

    protected   AttributeKey<String> _deviceId = AttributeKey.valueOf("deviceId");

    /**
     * 登录成功在当前连接中添加成功属性
     * @param ctx
     * @param deviceId
     */
    private void loginSuccess(ChannelHandlerContext ctx, String deviceId) {
        ctx.attr(_login).set(true);
        ctx.attr(_deviceId).set(deviceId);

        MqttChannelHandlerContext context = new MqttChannelHandlerContext(ctx);
        context.setLogin(true);
        context.setDeviceId(deviceId);

        NodeChannelHandlerContextCache.addMqttChannelHandlerContext(context);
    }


    private void connectBack(ChannelHandlerContext ctx,  MqttConnectReturnCode connectReturnCode) {
        MqttConnAckVariableHeader mqttConnAckVariableHeader = new MqttConnAckVariableHeader(connectReturnCode, true);

        MqttFixedHeader mqttFixedHeader = new MqttFixedHeader(
                MqttMessageType.CONNACK,false, MqttQoS.AT_MOST_ONCE, false, 0x02);
        MqttConnAckMessage connAck = new MqttConnAckMessage(mqttFixedHeader, mqttConnAckVariableHeader);
        ctx.writeAndFlush(connAck);
    }

    public void connectionTimeoutHandler(ChannelHandlerContext ctx, IdleStateEvent msg) {
        //会话超时，直接关闭连接
        ctx.close();
    }


    public void closeChannelHandlerContext(ChannelHandlerContext context) {
        MqttChannelHandlerContext ctx = getMqttChannelHandlerContext(context);
        if( ctx != null) {
            log.info("开始处理设备断线，设备ID={}", ctx.getDeviceId());
            if(!Strings.isNullOrEmpty(ctx.getDeviceId())) {
                this.willService.willHandler(ctx.getDeviceId());
            }

            this.mqttMessageService.closeConnect(ctx);
            closeLogin(ctx);
        }
    }
}